#include "Publisher.h"
#include "Broker.h"
#include <zmq.h>
#include <assert.h>
#include <string.h>


Publisher::Publisher():pub_(0)
{
  pub_ = zmq_socket(Broker::Context(), ZMQ_PUB);
  assert(pub_);

  const int rc=zmq_connect(pub_, Broker::XsubAddr);
  assert(rc==0);
}

Publisher::~Publisher()
{
  zmq_close(pub_);
}

void Publisher::publish(const ::google::protobuf::Message& msg)
{
//printf("(%s:%d) publishing '%s'\n",__FILE__,__LINE__,msg.GetTypeName().c_str());
  //send topic
  {
    zmq_msg_t m;
    const std::string topic=msg.GetTypeName()+'\0';
    const int len=topic.length();

    const int rc1 = zmq_msg_init_size(&m, len);
    assert(rc1==0);

    const int rc2 = zmq_msg_init_data(&m, (void*)topic.c_str(), len,0,0);
    assert(rc2==0);

    const int rc3 = zmq_sendmsg(pub_, &m, ZMQ_SNDMORE);
    assert(rc3==len);
  }

  //send content
  {
    zmq_msg_t m;
    const std::string content=msg.SerializeAsString();
    int rc = zmq_msg_init_size (&m, content.length());
    assert (rc == 0);
    const void* data=content.c_str();
    memcpy(zmq_msg_data(&m),data,content.length());
    rc = zmq_sendmsg (pub_, &m, 0);
    assert (rc == content.length());
  }

}

